package org.mq.mymq.broker.event.listener;

import com.alibaba.fastjson.JSON;
import org.apache.commons.collections4.CollectionUtils;
import org.mq.mymq.broker.cache.CommonCache;
import org.mq.mymq.broker.event.model.ConsumeMsgEvent;
import org.mq.mymq.broker.model.ConsumeQueueConsumeReqModel;
import org.mq.mymq.broker.rebalance.ConsumerInstance;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.ConsumeMsgBaseRespDTO;
import org.mq.mymq.common.dto.ConsumeMsgReqDTO;
import org.mq.mymq.common.dto.ConsumeMsgRespDTO;
import org.mq.mymq.common.enums.BrokerResponseCode;
import org.mq.mymq.common.event.Listener;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @ClassName ConsumeMsgListener
 * @Description
 * @Author jiarong_ye
 * @Date 2025/6/28 18:56
 * @Version 1.0
 */
public class ConsumeMsgListener implements Listener<ConsumeMsgEvent> {
    @Override
    public void onReceive(ConsumeMsgEvent event) throws Exception {
        ConsumeMsgReqDTO consumeMsgReqDTO = event.getConsumeMsgReqDTO();
        String currentReqId = consumeMsgReqDTO.getIp() + ":" + consumeMsgReqDTO.getPort();
        String topic = consumeMsgReqDTO.getTopic();
        ConsumerInstance consumerInstance = new ConsumerInstance();
        consumerInstance.setTopic(topic);
        consumerInstance.setIp(consumeMsgReqDTO.getIp());
        consumerInstance.setPort(consumeMsgReqDTO.getPort());
        consumerInstance.setConsumerRegId(consumeMsgReqDTO.getIp() + ":" + consumeMsgReqDTO.getPort());
        consumerInstance.setConsumeGroup(consumeMsgReqDTO.getConsumeGroup());
        consumerInstance.setBatchSize(consumeMsgReqDTO.getBatchSize());
        // 加入消费池
        CommonCache.getConsumerInstancePool().addInstancePool(consumerInstance);
        Map<String, List<ConsumerInstance>> consumerGroupMap = CommonCache.getConsumerHoldMap().get(topic);
        ConsumeMsgBaseRespDTO consumeMsgBaseRespDTO = new ConsumeMsgBaseRespDTO();
        List<ConsumeMsgRespDTO> consumeMsgRespDTOList = new ArrayList<>();
        consumeMsgBaseRespDTO.setConsumeMsgRespDTOList(consumeMsgRespDTOList);
        consumeMsgBaseRespDTO.setMsgId(event.getMsgId());
        if (consumerGroupMap == null) {
            // 消费组还没有经过重平衡，不会那么快消费到数据，所以要通知客户端目前服务端未分配队列给该消费组
            event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.CONSUME_MSG_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgBaseRespDTO)));
            return;
        }
        List<ConsumerInstance> consumerInstances = consumerGroupMap.get(consumeMsgReqDTO.getConsumeGroup());
        if (CollectionUtils.isEmpty(consumerInstances)) {
            event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.CONSUME_MSG_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgBaseRespDTO)));
            return;
        }
        for (ConsumerInstance instance : consumerInstances) {
            if (instance.getConsumerRegId().equals(currentReqId)) {
                // 当前消费者有占用队列的权利，可以消费 TODO
                for (Integer queueId : instance.getQueueIdSet()) {
                    ConsumeQueueConsumeReqModel consumeQueueConsumeReqModel = new ConsumeQueueConsumeReqModel();
                    consumeQueueConsumeReqModel.setTopic(topic);
                    consumeQueueConsumeReqModel.setConsumeGroup(instance.getConsumeGroup());
                    consumeQueueConsumeReqModel.setQueueId(queueId);
                    consumeQueueConsumeReqModel.setBatchSize(instance.getBatchSize());
                    List<byte[]> commitLogContentList = CommonCache.getConsumeQueueConsumeHandler().consume(consumeQueueConsumeReqModel);
                    ConsumeMsgRespDTO consumeMsgRespDTO = new ConsumeMsgRespDTO();
                    consumeMsgRespDTO.setQueueId(queueId);
                    consumeMsgRespDTO.setCommitLogContentList(commitLogContentList);
                    consumeMsgRespDTOList.add(consumeMsgRespDTO);
                }
            }
        }
        // 直接返回空数据
        event.getChannelHandlerContext().writeAndFlush(new TcpMsg(BrokerResponseCode.CONSUME_MSG_RESP.getCode(), JSON.toJSONBytes(consumeMsgBaseRespDTO)));
    }
}
